package com.hsc.www.reSend.core;

import com.alibaba.fastjson.JSONObject;
import com.hsc.www.reSend.core.delayedQueue.CapillaryDelayedQueue;
import com.hsc.www.reSend.core.delayedQueue.CapillaryDelayedQueueFactory;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.io.IOException;
import java.util.*;

@Component
public class UserReceiver {
    private final CapillaryDelayedQueueFactory capillaryDelayedQueueFactory;

    public UserReceiver(CapillaryDelayedQueueFactory capillaryDelayedQueueFactory) {
        this.capillaryDelayedQueueFactory = capillaryDelayedQueueFactory;
    }


    @RabbitListener(queues = Config.DELAYED_MESSAGE_REGISTRAR_QUEUE)
    public void handleRegistrarDelayedMessage(Channel channel, Message message) throws Exception {
        System.out.println("延迟消息：" + JSONObject.toJSONString(message));

        DelayedMessageProperties delayedMessageProperties = this.getDelayedMessage(message);
        if (Objects.isNull(delayedMessageProperties)) {
            this.handleUndeliverableMessage(channel, message);
            return;
        }

        int doDelayedTime;
        int delayedMsgDelayedTime = ObjectUtils.defaultIfNull(delayedMessageProperties.getDelayedMsgDelayedTime(), 0);
        boolean delayedMsgAlreadyDelay = ObjectUtils.defaultIfNull(delayedMessageProperties.getDelayedMsgAlreadyDelay(), false);
        if (delayedMsgDelayedTime > 0 && !delayedMsgAlreadyDelay) {//处理未进行延迟的延迟消息
            doDelayedTime = delayedMsgDelayedTime;
            delayedMessageProperties.setDelayedMsgAlreadyDelay(true);
        } else {
            Integer totalAttempts = ObjectUtils.defaultIfNull(delayedMessageProperties.getTotalTries(), 0);
            Integer alreadyAttempts = ObjectUtils.defaultIfNull(delayedMessageProperties.getAlreadyTries(), 0);
            if (alreadyAttempts >= totalAttempts) {
                this.handleDeathMessage(channel, message, delayedMessageProperties);
                return;
            }

            List<Integer> tryDelayedTimeList = ObjectUtils.defaultIfNull(delayedMessageProperties.getTryDelayedTimeList(), Collections.EMPTY_LIST);
            doDelayedTime = alreadyAttempts < tryDelayedTimeList.size() ?
                    tryDelayedTimeList.get(alreadyAttempts)
                    : (tryDelayedTimeList.isEmpty() ? 0 : tryDelayedTimeList.get(tryDelayedTimeList.size() - 1));
            delayedMessageProperties.setAlreadyTries(alreadyAttempts + 1);
        }
        

        CapillaryDelayedQueue capillaryDelayedQueue = capillaryDelayedQueueFactory.pick(doDelayedTime);
        Assert.isTrue(capillaryDelayedQueue != null, "未找到有效的延迟队列 延迟时间=" + doDelayedTime);

        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder()
                .headers(new HashMap() {{
                    put(DelayedMessageProperties.X_ORIGINAL_QUEUE_INFO, JSONObject.toJSONString(delayedMessageProperties));
                }})
                .build();
        try {
            channel.txSelect();
            channel.basicPublish("", capillaryDelayedQueue.getDelayedQueueName(), true, basicProperties, message.getBody());
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            channel.txCommit();
        } catch (Exception e) {
            channel.txRollback();
        }

    }


    @RabbitListener(queues = Config.DELAYED_MESSAGE_RELIVE_QUEUE)
    public void handleReliveDelayedMessage(Channel channel, Message message) throws Exception {
        System.out.println("重新分发：" + JSONObject.toJSONString(message));

        DelayedMessageProperties delayedMessageProperties = this.getDelayedMessage(message);
        assert delayedMessageProperties != null;
        FreeBasicProperties freeBasicProperties = delayedMessageProperties.getBasicProperties();
        freeBasicProperties.getHeaders().put(DelayedMessageProperties.X_ORIGINAL_QUEUE_INFO, JSONObject.toJSONString(delayedMessageProperties));

        channel.basicPublish(delayedMessageProperties.getExchange(), delayedMessageProperties.getRoutingKey(), true, freeBasicProperties.convert2AmqpBasicProperties(), message.getBody());
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


    private DelayedMessageProperties getDelayedMessage(Message message) {

        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        if (Objects.isNull(headers)) {
            return null;
        }

        Object originalDelayedMessage = headers.get(DelayedMessageProperties.X_ORIGINAL_QUEUE_INFO);
        if (Objects.isNull(originalDelayedMessage)) {
            return null;
        }
        return JSONObject.parseObject(originalDelayedMessage.toString(), DelayedMessageProperties.class);
    }

    /**
     * 处理无法投递的消息
     */
    private void handleUndeliverableMessage(Channel channel, Message message) throws IOException {
        System.out.println("无法投递消息：" + JSONObject.toJSONString(message));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

    /**
     * 处理信息死亡的消息
     */
    private void handleDeathMessage(Channel channel, Message message, DelayedMessageProperties delayedMessageProperties) throws IOException {
        System.out.println("死亡消息：" + JSONObject.toJSONString(message));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


}
